package org.lql.chaos.mq.rabbitmq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import java.nio.charset.StandardCharsets;

/**
 * @author: liangqinglong
 * @date: 2025-08-08 10:56
 * @description:
 **/
@Slf4j
@Configuration
@EnableRabbit
public class RabbitConfig {

	private final ConnectionFactory connectionFactory;

	public RabbitConfig(ConnectionFactory connectionFactory) {
		this.connectionFactory = connectionFactory;
	}

	//进行rabbitmq template的配置
	@Bean(name = "rabbitTemplate")
	@Primary
	public RabbitTemplate rabbitTemplate() {
		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
		rabbitTemplate.setEncoding("UTF-8");
		rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
		// 如果消息发送到交换机后，根据路由键无法找到匹配的队列，消息将被返回给生产者
		// 配合 setReturnsCallback() 回调使用，可以捕获并处理这些无法路由的消息
		// 如果设置为 false（默认值），无法路由的消息会被直接丢弃
		rabbitTemplate.setMandatory(true);
		// 消息确认机制，确认消息是否被Broker成功接收
		rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
			String messageId = (correlationData != null) ? correlationData.getId() : "unknown";
			if (ack) {
				log.info("Publisher confirm ACK, id={}", messageId);
			} else {
				log.error("Publisher confirm NACK, id={}, cause={}", messageId, cause);
			}
		});
		// 当消息发送到交换机后，但是没有路由到队列时，会触发该回调
		rabbitTemplate.setReturnsCallback(ret -> {
			String body = new String(ret.getMessage().getBody(), StandardCharsets.UTF_8);
			String messageId = ret.getMessage().getMessageProperties().getMessageId();
			log.error("消息未路由: code={}, text={}, exchange={}, rk={}, msgId={}, body={}",
					ret.getReplyCode(), ret.getReplyText(), ret.getExchange(), ret.getRoutingKey(), messageId, body);
			// 可以进行重发或者记录日志、触发告警邮件、短信等
		});
		return rabbitTemplate;
	}

	@Bean
	public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory);

		Jackson2JsonMessageConverter json = new Jackson2JsonMessageConverter();
		SimpleMessageConverter stringConv = new SimpleMessageConverter();

		ContentTypeDelegatingMessageConverter composite = new ContentTypeDelegatingMessageConverter(json); // ★ 默认 JSON
		composite.addDelegate("application/json", json);
		composite.addDelegate("text/plain", stringConv);
		composite.addDelegate("text/x-json", json);

		factory.setMessageConverter(composite);
		factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

		return factory;
	}

}
